package com.cloudfun.msglisten.ws.haier.handler;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONObject;
import com.cloudfun.msglisten.common.constant.RedisKey;
import com.cloudfun.msglisten.entity.HaierMessage;
import com.cloudfun.msglisten.service.HaierMessageService;
import com.cloudfun.msglisten.util.BloomFilterHelper;
import com.cloudfun.msglisten.util.RedisBloomFilter;
import com.cloudfun.msglisten.ws.haier.dto.MessageDTO;
import com.google.common.hash.Funnels;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.BooleanUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;
import java.util.List;
import java.util.Set;

/**
 * Created by Huqin on 2019/12/17
 */
@Slf4j
@Component
public class HaierWsDeviceMsgHandler implements HaierWsMsgHandler, InitializingBean {

    @Autowired
    private HaierMessageService haierMessageService;

    @Autowired
    private RedisBloomFilter redisBloomFilter;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private List<DeviceMsgHandler> deviceMsgHandlers;

    private BloomFilterHelper<CharSequence> bloomFilterHelper;

    private static final int BLOOM_FILTER_MIN_ELEMENTS = 100_000;

    @Override
    public void afterPropertiesSet() throws Exception {
        bloomFilterHelper = new BloomFilterHelper<>(Funnels.stringFunnel(Charset.defaultCharset()), BLOOM_FILTER_MIN_ELEMENTS, 0.1D);
    }

    /**
     * 根据Imei集合更新布隆过滤器
     */
    public void updateImeiBloomFilter() {
        Set<String> imeiSet = stringRedisTemplate.opsForSet().members(RedisKey.IMEI_SET);
        if (CollectionUtil.isEmpty(imeiSet)) {
            return;
        }
        for (String imei : imeiSet) {
            // 布隆过滤器不包含的话一定不包含
            if (!redisBloomFilter.includeByBloomFilter(bloomFilterHelper, RedisKey.REDIS_BLOOM_FILTER_KEY, imei)) {
                redisBloomFilter.addByBloomFilter(bloomFilterHelper, RedisKey.REDIS_BLOOM_FILTER_KEY, imei);
            }
        }
    }

    /**
     * 添加设备号到布隆过滤器
     *
     * @param imei
     */
    public void addImeiToBloomFilter(String imei) {
        // 添加imei集合
        stringRedisTemplate.opsForSet().add(RedisKey.IMEI_SET, imei);
        // 添加到布隆过滤器
        redisBloomFilter.addByBloomFilter(bloomFilterHelper, RedisKey.REDIS_BLOOM_FILTER_KEY, imei);
    }

    @Override
    public void handle(String message) {
        // filter
        MessageDTO messageDTO = JSONObject.parseObject(message, MessageDTO.class);
        String imei = messageDTO.getHeader().getKeys().getImei();
        boolean include = redisBloomFilter.includeByBloomFilter(bloomFilterHelper, RedisKey.REDIS_BLOOM_FILTER_KEY, imei);
        if (!include || BooleanUtils.isFalse(stringRedisTemplate.opsForSet().isMember(RedisKey.IMEI_SET, imei))) {
            // 如果布隆过滤器中没有，或者布隆过滤器中有，但是数据库中没有，则直接返回
            return;
        }
        log.debug("收到消息: {}", message);
        // save
        HaierMessage haierMessage = HaierMessage.builder()
                .typeId(messageDTO.getHeader().getKeys().getTypeId())
                .imei(imei)
                .timestamp(messageDTO.getHeader().getTimestamp().toString())
                .name(messageDTO.getBody().getName())
                .args(messageDTO.getBody().getArgs())
                .build();
        haierMessageService.create(haierMessage);
        // handle
        boolean handled = false;
        for (DeviceMsgHandler deviceMsgHandler : deviceMsgHandlers) {
            if (deviceMsgHandler.canHandle(haierMessage.getName())) {
                deviceMsgHandler.handle(haierMessage);
                handled = true;
            }
        }
        if (!handled) {
            log.error("消息无法处理，topic：{}，messageId：{}", haierMessage.getName(), haierMessage.getId());
        }
    }

}
